package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * 代码性质:
 *   1. 插件性质
 *      1) 实现接口或者继承类
 *      2) 实现抽象方法或者重写具体方法
 *      3) 打包扔到集群运行
 *   2. 客户端性质
 *      1) 创建客户端对象
 *      2) 调用方法完成功能
 *      3) 关闭对象
 *
 *
 * 生产者 - 异步发送 - 不带回调
 */
public class ProducerDemo {
    public static void main(String[] args) {
        //1. 创建生产者对象
        Properties properties = new Properties();
        // 指定kafka集群的位置
        properties.put("bootstrap.servers", "hadoop102:9092,hadoop103:9092,hadoop104:9092");
        // 批次大小 默认16K
        properties.put("batch.size", 16384);
        // 等待时间
        properties.put("linger.ms", 1);
        // RecordAccumulator缓冲区大小 默认32M
        properties.put("buffer.memory", 33554432);

        // key,value序列化
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");


        // 泛型: K ->  消息的key的类型(key可有可无) , V -> 消息的类型
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        //2. 调用方法，发送消息
        for (int i = 0; i < 10000 ; i++) {
            producer.send(new ProducerRecord<String,String>("first", "shangguigu==>" + i ));
        }

        //3. 关闭生产者对象
        producer.close();
    }

}
